package com.you.raft.core.impl;

import com.you.raft.common.entity.*;
import com.you.raft.core.ClientService;
import com.you.raft.core.LogModule;
import com.you.raft.core.StateMachine;
import com.you.raft.rpc.RpcRaftClient;
import com.you.raft.common.entity.Request;
import com.you.raft.thread.RaftThreadPool;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 客户端服务
 * 复制将客户端对状态操作的请求转发至Leader节点,并处理请求
 */
@Slf4j
public class DefaultClientService implements ClientService {
    private final DefaultRaftNode node;
    private final LogModule logModule;
    private final StateMachine stateMachine;
    private final RpcRaftClient rpcClient;
    private final Cluster cluster;


    public DefaultClientService(DefaultRaftNode node) {
        this.node = node;
        this.logModule = node.getLogModule();
        this.stateMachine = node.getStateMachine();
        this.rpcClient = node.getRpcClient();
        this.cluster = node.getCluster();
    }

    @Override
    public synchronized ClientResponse clientRequest(ClientRequest request) {
        log.info("客户端请求命令:[ {} ], key : [{}], value : [{}]",
                ClientRequest.Type.value(request.getType()), request.getKey(), request.getValue());
        if (node.getStatus() != NodeStatus.LEADER) {
            log.info("当前节点不是Leader节点,转发请求从{} 到: {}",
                    cluster.getSelf().getAddress(), cluster.getLeader().getAddress());
            return redirect(request);
        }
        if (request.getType() == ClientRequest.GET) {
            String value = stateMachine.get(request.getKey());
            if (value != null) {
                return new ClientResponse(value);
            }
            return ClientResponse.fail();
        }
        //创建日志,日志索引会在write()方法里面正确设置
        LogEntry logEntry = LogEntry.builder()
                .command(Command.builder().key(request.getKey()).value(request.getValue()).build())
                .term(node.getCurrentTerm())
                .index(0).build();
        logModule.write(logEntry);
        log.info("创建日志 : {}, 日志索引 : {}", logEntry, logEntry.getIndex());

        final AtomicInteger sucCount = new AtomicInteger(0);
        //  复制到其他机器
        List<Future<Boolean>> taskList = new ArrayList<>();
        for (Node peer : cluster.getOtherNodes()) {
            taskList.add(node.replication(peer, logEntry));
        }
        int taskCount = taskList.size();
        CountDownLatch latch = new CountDownLatch(taskCount);
        for (Future<Boolean> future : taskList) {
            RaftThreadPool.execute(() -> {
                try {
                    if (future.get(3000, TimeUnit.MILLISECONDS)) {
                        sucCount.incrementAndGet();
                    }
                } catch (Exception e) {
                    log.error("错误", e);
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await(4000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.error("错误", e);
        }
        //求matchIndexs已经提交日志的中位数，如果中位数>=commitIndex,说明绝大多数节点已经提交了
        List<Long> matchIndexList = new ArrayList<>(node.getMatchIndexs().values());
        int median = 0;
        if (matchIndexList.size() >= 2) {
            Collections.sort(matchIndexList);
            median = (matchIndexList.size() + 1) / 2;
        }
        long N = matchIndexList.get(median);
        if (N >= node.getCommitIndex()) {
            LogEntry entry = logModule.get(N);
            if (entry != null && entry.getTerm() == node.getCurrentTerm()) {
                node.setCommitIndex(N);
            }
        }
        //  响应客户端(成功一半)-则应用到状态机
        if (sucCount.get() >= (taskCount + 1) / 2) {
            node.setCommitIndex(logEntry.getIndex());
            stateMachine.apply(logEntry);
            node.setLastApplied(node.getCommitIndex());
            log.info("日志[ {} ]成功应用到状态机", logEntry);
            return ClientResponse.ok();
        } else {
            // 回滚已经提交的日志.
            logModule.removeStartWith(logEntry.getIndex());
            log.warn("回滚日志: {}", logEntry);
            return ClientResponse.fail();
        }
    }

    public ClientResponse redirect(ClientRequest request) {
        Request r = Request.builder()
                .obj(request)
                .addr(cluster.getLeader().getAddress())
                .cmd(Request.CLIENT_ORDER).build();
        log.warn("{}", rpcClient);
        Object result = rpcClient.send(r);
        log.warn("{}", result);
        return (ClientResponse) result;
    }

}
